#include "cppthreadpool.h"
#include <string.h>
#include <unistd.h>
#include <string>

TaskQue::TaskQue()
{
    pthread_mutex_init(&mutexQue,NULL);
}

TaskQue::~TaskQue()
{
    pthread_mutex_destroy(&mutexQue);
}

void TaskQue::addTask(Task& task) //添加任务
{
    pthread_mutex_lock(&mutexQue);
    mTaskQ.push(task);
    pthread_mutex_unlock(&mutexQue);
}

void TaskQue::addTask(callback func, void* arg) //添加任务
{
    pthread_mutex_lock(&mutexQue);
    Task t;
    t.function = func;
    t.arg = arg;
    mTaskQ.push(t);
    pthread_mutex_unlock(&mutexQue);
}

Task TaskQue::execTask() 
{
    Task t;
    pthread_mutex_lock(&mutexQue);
    if(mTaskQ.size() > 0)
    {
        t = mTaskQ.front();
        mTaskQ.pop();
    }  
    pthread_mutex_unlock(&mutexQue);
    return t;
}

CppThreadPool::CppThreadPool(int min, int max)
{
    //init 
    do
    {
        mQue = new TaskQue;
        if(mQue == nullptr)
        {
            cout << "new mque error\n";
            break;  
        }

        mThreadIDs = new pthread_t[max];
        if(mThreadIDs == nullptr)
        {
            cout << "iids malloc error\n";
            break;
        }
        memset(mThreadIDs, 0, sizeof(pthread_t) * max);

        mMinNum = min; 
        mMaxNum = max; 
        mLiveNum = min;
        mBusyNum = 0;
        mExitNum = 0;
        mShutDown = false;

        if(pthread_mutex_init(&mMutexPool, NULL) != 0 ||
           pthread_cond_init(&mEmpty, NULL) != 0 )
        {
            cout << "mutex or cond init error\n ";
            break;
        }

        //创建线程
        //管理者线程
        pthread_create(&mManagerID, NULL, manager, this);
        //创建工作线程
        for (size_t i = 0; i < min; i++)
        {
            pthread_create(&mThreadIDs[i], NULL, worker, this);
            cout << "创建子线程, ID: " << to_string(mThreadIDs[i]) << endl;
        }
        return;
    } while (0);

    //释放资源
    if(mThreadIDs)
    {
        delete [] mThreadIDs;
        mThreadIDs = nullptr;
    }
    if(mQue)
    {
        delete mQue;
        mQue == nullptr;
    }
}

CppThreadPool::~CppThreadPool()
{
    //关闭线程池
    mShutDown = true;

    //阻塞回收管理者线程
    pthread_join(mManagerID,NULL);

    //唤醒阻塞的消费者线程
    for (size_t i = 0; i < mLiveNum; i++)
    {
        pthread_cond_signal(&mEmpty);
    }

    //释放堆内存  
    if(mThreadIDs)
    {
        delete [] mThreadIDs;
        mThreadIDs = nullptr;
    }    
    if(mQue)
    {
        delete mQue;
        mQue = nullptr;
    }

    //释放互斥锁和条件变量
    pthread_cond_destroy(&mEmpty);
    pthread_mutex_destroy(&mMutexPool);
}

void CppThreadPool::addTaskToPool(Task task) //添加任务
{
    if(mShutDown)
        return;
    mQue->addTask(task);
    pthread_cond_signal(&mEmpty);
}

int CppThreadPool::getLiveNum()//获取工作的线程个数
{
     pthread_mutex_lock(&mMutexPool);
     int _num = mLiveNum;
     pthread_mutex_unlock(&mMutexPool);
     return _num;
}

int CppThreadPool::getBusyNum()//获取工作的busy的线程个数
{
     pthread_mutex_lock(&mMutexPool);
     int _num = mBusyNum;
     pthread_mutex_unlock(&mMutexPool);
     return _num;
}

void* CppThreadPool::worker(void* arg)
{
    CppThreadPool* pool = static_cast<CppThreadPool*>(arg);
    // 一直不停的工作
    while (true)
    {
        // 访问任务队列(共享资源)加锁
        pthread_mutex_lock(&pool->mMutexPool);
        // 判断任务队列是否为空, 如果为空工作线程阻塞
        while (pool->mQue->getmTaskQNum() == 0 && !pool->mShutDown)
        {
            cout << "thread " << to_string(pthread_self()) << " waiting..." << endl;
            // 阻塞线程
            pthread_cond_wait(&pool->mEmpty, &pool->mMutexPool);

            // 解除阻塞之后, 判断是否要销毁线程
            if (pool->mExitNum > 0)
            {
                pool->mExitNum--;
                if (pool->mLiveNum > pool->mMinNum)
                {
                    pool->mLiveNum--;
                    pthread_mutex_unlock(&pool->mMutexPool);
                    pool->threadExit();
                }
            }
        }
        // 判断线程池是否被关闭了
        if (pool->mShutDown)
        {
            pthread_mutex_unlock(&pool->mMutexPool);
            pool->threadExit();
        }

        // 从任务队列中取出一个任务
        Task task = pool->mQue->execTask();
        // 工作的线程+1
        pool->mBusyNum++;
        // 线程池解锁
        pthread_mutex_unlock(&pool->mMutexPool);
        // 执行任务
        cout << "thread " << to_string(pthread_self()) << " start working..." << endl;
        task.function(task.arg);
        delete task.arg;
        task.arg = nullptr;

        // 任务处理结束
        cout << "thread " << to_string(pthread_self()) << " end working...\n";
        pthread_mutex_lock(&pool->mMutexPool);
        pool->mBusyNum--;
        pthread_mutex_unlock(&pool->mMutexPool);
    }
    return nullptr;
}


void* CppThreadPool::manager(void* arg)
{
    CppThreadPool* pool = static_cast<CppThreadPool*>(arg);

    while (!pool->mShutDown)
    {
        //每间隔3秒检测一次
        sleep(3);

        //取出当前线程池中任务数量和当前线程数量
        //取出忙的线程个数
        pthread_mutex_lock(&pool->mMutexPool);
        int queSize = pool->mQue->getmTaskQNum();
        int liveNum = pool->mLiveNum;
        int busyNUm = pool->mBusyNum;
        pthread_mutex_unlock(&pool->mMutexPool);

        //添加线程
        //任务个数 > 存活个数 && 存活个数 < 最大线程
        if(queSize > liveNum && liveNum < pool->mMaxNum)
        {
            pthread_mutex_lock(&pool->mMutexPool);
            int counter = 0;
            for (size_t i = 0; 
                 i < pool->mMaxNum && counter < COUNTER 
                     && pool->mLiveNum < pool->mMaxNum;
                 i++)
            {
                if(pool->mThreadIDs[i] == 0)
                {
                    pthread_create(&pool->mThreadIDs[i], NULL, worker, pool);
                    counter++;
                    pool->mLiveNum++;
                }
            }       
            pthread_mutex_unlock(&pool->mMutexPool);
        }
        //销毁线程
        //忙的线程*2  < 存活的线程数 && 存活的线程 > 最小的线程数
        if(busyNUm* 2 < liveNum && liveNum > pool->mMinNum)
        {
            pthread_mutex_lock(&pool->mMutexPool);
            pool->mExitNum = COUNTER;
            pthread_mutex_unlock(&pool->mMutexPool);

            //让线程自杀
            for (size_t i = 0; i < COUNTER; i++)
            {
                pthread_cond_signal(&pool->mEmpty);
            }
        }
    }
}

void CppThreadPool::threadExit()
{
    pthread_t pid = pthread_self();
    for (size_t i = 0; i < mMaxNum; i++)
    {
        if(mThreadIDs[i] == pid)
        {
            mThreadIDs[i] = 0;
            cout << "threadexit() call," << pid << "exiting\n";
            break;
        }
    }
    pthread_exit(NULL); 
}

